package net.bwie.realtime.dws.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.dws.log.bean.AreaControl;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;

public class TrafficCarMapFunction extends RichMapFunction<String, AreaControl> {

    //step1.声明状态
    private transient ValueState<String> car = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        //step2.初始化状态
        ValueStateDescriptor<String> descriptor = new ValueStateDescriptor<>("car", String.class);

        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.days(1))
                .updateTtlOnCreateAndWrite()
                .neverReturnExpired()
                .build();
        descriptor.enableTimeToLive(ttlConfig);

        car = getRuntimeContext().getState(descriptor);
    }

    @Override
    public AreaControl map(String value) throws Exception {


        JSONObject jsonObject = JSON.parseObject(value);
        String car1 = jsonObject.getString("car");

        //车辆数
        Long carCount = 0L;
        if (null == this.car.value()){
            //tep3.更新状态
            carCount = 1L;
            this.car.update(car1);
        }else{
            if (!this.car.value().equals(car1)){
                carCount = 1L;
                this.car.update(car1);
            }
        }

        return new AreaControl(
                null,
                null,
                jsonObject.getString("areaId"),
                carCount,
                jsonObject.getLong("actionTime")
        );
    }
}
